"""
@Author    : ghenyar
@Time      : 2025/8/28 12:32
@File      : tools
@Desc      : 工具函数
"""
import hashlib
import random
import re
import string

import pandas as pd
from datetime import datetime
from typing import Optional, List, Union, Any
from fastapi import Request
from passlib.context import CryptContext
from sqlalchemy import Sequence


def format_datetime(date: Optional[datetime | None]) -> Optional[str]:
    """
    时间格式化处理
    :param date: 时间
    :return:
    """
    if date is None or date == "":
        return None
    if isinstance(date, str):
        try:
            date = datetime.fromisoformat(date)
        except ValueError:
            return None
    return date.strftime("%Y-%m-%d %H:%M:%S")


def find_children(data_list, parent_id, parent_field: str = "p_id"):
    """
    查找子级数据
    :param data_list: 数据
    :param parent_id: 父级
    :param parent_field: 父级字段
    :return:
    """
    children = []
    for obj in data_list:
        if getattr(obj, parent_field) == parent_id:
            children.append(obj)
    return children


def select_orm(data_list: List[dict:[any]], label: str, value: str):
    """
    下拉框格式数据序列化
    :param data_list:  数据
    :param label:  下拉框label显示的数据字段
    :param value:  下拉框value显示的数据字段
    :return:
    """
    return [{"label": getattr(data, label), "value": getattr(data, value)} for data in data_list]


def select_orm_key(data_list: List[dict:[any]], label: str, key: str):
    """
    下拉框格式数据序列化
    :param data_list:  数据
    :param label:  下拉框label显示的数据字段
    :param key:  下拉框value显示的数据字段
    :return:
    """
    return [{"label": getattr(data, label), "key": getattr(data, key)} for data in data_list]


def select_tree_orm(data_list: List[Any] | Sequence, label: str, key: str, p_field: str = "p_id"):
    """
    树形下拉框
    :param data_list:
    :param label:
    :param key:
    :param p_field:
    :return:
    """
    top_level = find_children(data_list, 0, p_field)
    # 递归构建整个树型选择结构
    return [select_tree(obj, data_list, label, key, p_field) for obj in top_level]


def select_tree(data: dict, data_list: List[dict], label: str, key: str, p_field: str):
    """树形下拉框的递归函数"""
    tree = {
        "label": getattr(data, label),
        "key": getattr(data, key),
    }
    children = [
        select_tree(child, data_list, label, key, p_field) for child in find_children(data_list, data.id, p_field)
    ]
    if children:
        tree["children"] = children
    return tree


def build_tree(data_list: Union[List[dict], tuple[Any], List[Any]], parent_id: int = 0,
               parent_field: str = "p_id"):
    """
    生成树形结构
    :param data_list: 数据
    :param parent_id: 上级id
    :param parent_field: 上级字段
    :return:
    """
    tree = []
    for item in data_list:
        # 数据处理
        if isinstance(item, dict):
            item_parent_id = item.get(parent_field)
            item_id = item.get("id")
        else:
            item_parent_id = getattr(item, parent_field, None)
            item_id = getattr(item, "id", None)
        item = {k: v for k, v in item.__dict__.items() if not k.startswith('_')}

        if item_parent_id == parent_id:
            # 使用递归函数获取子节点
            children = build_tree(data_list, item_id, parent_field)
            if children:
                item["children"] = children
            tree.append(item)
    return tree


def audio_video_time_format(time_str: str, abandon_zero: Optional[bool] = True):
    """
    音频、视频时长格式化
    :param time_str: 时长
    :param abandon_zero: 是否舍去0  True：是  False：否    如时长是：0:21:55(时:分:秒) True则舍去0 变成：21:55(分:秒)
    :return:
    """
    # 将时间字符串转换为datetime对象
    time_obj = datetime.strptime(time_str, "%H:%M:%S.%f")
    # 将时间对象的秒数四舍五入到整数
    rounded_seconds = round(time_obj.second + time_obj.microsecond / 1000000)
    if abandon_zero:
        # 根据小时部分是否为0来决定输出格式
        if time_obj.hour == 0 or time_obj.hour == 00:
            if time_obj.minute == 0 or time_obj.minute == 00:
                formatted_time = f"{time_obj.second}"
            else:
                formatted_time = f"{time_obj.minute}:{time_obj.second}"
        else:
            formatted_time = f"{time_obj.hour}:{time_obj.minute}:{rounded_seconds}"
    else:
        formatted_time = f"{time_obj.hour}:{time_obj.minute}:{rounded_seconds}"
    return formatted_time


def get_protocol(request: Request) -> str:
    """
    从请求中判断并返回协议类型（http 或 https）。
    :param request: FastAPI 的 Request 对象。
    :return: "https" 或 "http"。
    """
    # request.url.scheme 是 Starlette 提供的获取协议方案的可靠方法
    return "https" if request.url.scheme == "https" else "http"


def get_domain(request: Request) -> str:
    """
    从请求中提取并构建完整的域名（协议 + 主机名 + 端口）。
    :param request: FastAPI 的 Request 对象。
    :return: 完整的域名，例如 "http://example.com:8000"。
    """
    scheme = request.url.scheme
    hostname = request.url.hostname
    # 仅在非标准端口时添加端口号
    port = f":{request.url.port}" if request.url.port else ""
    return f"{scheme}://{hostname}{port}"


def generate_random_digits(length: int) -> str:
    """
    生成指定长度的纯数字随机字符串。
    :param length: 随机字符串的长度。
    :return: 生成的随机数字字符串。
    """
    return "".join(random.choices(string.digits, k=length))


def generate_random_string(length: int) -> str:
    """
    生成指定长度的字母和数字混合的随机字符串。
    :param length: 随机字符串的长度。
    :return: 生成的随机字符串。
    """
    return "".join(random.choices(string.ascii_letters + string.digits, k=length))


def get_ip(request: Request) -> str:
    """
    获取客户端的真实 IP 地址。
    会检查 'X-Forwarded-For' 和 'X-Real-IP' 这两个常见的代理头。
    :param request: FastAPI 的 Request 对象。
    :return: 客户端的 IP 地址字符串。
    """
    # 'X-Forwarded-For' 是一个 HTTP 扩展头部，用来表示 HTTP 请求端真实 IP
    x_forwarded_for = request.headers.get("X-Forwarded-For")
    if x_forwarded_for:
        # 如果经过多个代理，'X-Forwarded-For' 的值可能是 "client, proxy1, proxy2"，第一个为真实 IP
        return x_forwarded_for.split(",")[0].strip()

    # 'X-Real-IP' 是另一个获取真实 IP 的头部，通常由 Nginx 等代理服务器设置
    x_real_ip = request.headers.get("X-Real-IP")
    if x_real_ip:
        return x_real_ip

    # 如果以上头部都不存在，则返回直连的客户端 IP
    return request.client.host


def create_order_sn(prefix: str) -> str:
    """
    生成唯一的订单号。
    格式为：前缀 + 年月日时分秒 + 4位随机数。
    :param prefix: 订单号的前缀，例如 "PAY"。
    :return: 生成的订单号字符串。
    """
    now = datetime.now()
    random_number_string = ''.join(str(random.randint(0, 9)) for _ in range(4))
    order_sn = prefix + now.strftime("%Y%m%d%H%M%S") + str(random_number_string)
    return order_sn


def is_url(path: str) -> bool:
    """
    使用正则表达式判断一个字符串是否是有效的 URL。
    :param path: 要验证的字符串。
    :return: 如果是有效 URL，返回 True，否则返回 False。
    """
    url_pattern = re.compile(
        r'^(https?|ftp)://'
        r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|'
        r'localhost|'
        r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}|'
        r'\[?[A-F0-9]*:[A-F0-9:]+\]?)'
        r'(?::\d+)?'
        r'(?:/?|[/?]\S+)$', re.IGNORECASE)

    return re.match(url_pattern, path) is not None


def extract_location(text: str) -> Optional[str]:
    """
    从文本中提取中文括号（）内的内容。
    :param text: 包含括号的文本。
    :return: 括号内的内容，如果找不到则返回 None。
    """
    match = re.search(r'（(.+?)）', text)
    if match:
        return match.group(1)
    return None


def calculate_growth_percentage(last_count: int, current_count: int) -> float:
    """
    计算两个数值之间的增长百分比。
    :param last_count: 上一期的数值。
    :param current_count: 当前期的数值。
    :return:
    """
    if last_count == 0:
        # 如果上期为 0，当期大于 0 则视为 100% 增长，否则无增长
        return 100.0 if current_count > 0 else 0.0
    change = current_count - last_count
    return (change / last_count) * 100


def format_number(num: int) -> str:
    """
     将大于等于 1000 的数字格式化为 'k' 的形式（例如 1200 -> '1.2k'）。
    :param num: 待格式化的整数。
    :return: 格式化后的字符串。
    """
    if num >= 1000:
        return f"{num / 1000:.1f}k"
    return str(num)


def safe_to_int(value: Any, default: int = 0) -> int:
    """
    安全地将值转换为整数。能处理 None, NaN, 空字符串等情况。
    :param value: 何可能的值。
    :param default: 转换失败时返回的默认值。
    :return: 转换后的整数或默认值。
    """
    if pd.isna(value):  # 处理 NaN
        return default
    try:
        # 尝试将值转换为整数
        num = float(str(value).strip())  # 先转字符串再转浮点数，处理可能的空格
        return int(num) if num.is_integer() else default
    except (ValueError, TypeError, AttributeError):
        return default


def safe_get(item: dict, key: str) -> Any:
    """
     从字典中安全地获取值。如果键不存在或值为 NaN，返回 None。
    :param item: 字典对象。
    :param key: 要获取的键。
    :return: 键对应的值，或 None。
    """
    value = item.get(key)
    return None if pd.isna(value) else value


def verify_password(plain_password, hashed_password):
    """
    验证密码
    :param plain_password: 明文密码
    :param hashed_password: 加密的密码
    :return: bool
    """
    try:
        pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
        return pwd_context.verify(plain_password, hashed_password)
    except Exception:
        return False


def get_password_hash(pw: str) -> str:
    """
    加密密码
    :param pw: 原始密码
    :return: 加密的密码
    """
    pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
    return pwd_context.hash(pw)
